
#include "unix_socket.h"
#include "luas.h"
#include "pkt.h"
#include "luas.h"
#define MAX_UNIX_SOCKET 64

static UnixSocket s_sockets[MAX_UNIX_SOCKET];

#define fd2unix_socket(fd)  (&s_sockets[fd])


static int unix_socket_new()
{
    int fd = 0;
    int i;
    for(i = 1; i < MAX_UNIX_SOCKET; i++)
    {
        if(s_sockets[i].used == 0)
        {
            fd = i;
            //初始化port
            bzero(&s_sockets[i], sizeof(UnixSocket));
            s_sockets[i].used = 1;
            break;
        }
    }
    js_sys("unix_socket", "new unix socket:%d\n", fd);
    return fd;
}
static int unix_socket_free(int fd){
	UnixSocket *unix_socket = fd2unix_socket(fd);
	unix_socket->used = 0;
}

static void setnonblock(int sockfd)
{
  int flags;

  flags = fcntl(sockfd, F_GETFL);
  flags |= O_NONBLOCK;
  fcntl(sockfd, F_SETFL, flags);
}

static void unix_socket_on_write(int sockfd, short event, void *args)
{
    int i, j;
    char* rbuf;
    int   rlen;
    int fd = (int)args;
    UnixSocket *unix_socket = fd2unix_socket(fd);
    int buffd = unix_socket->buf_write;
    //分两段写
    for(i = 0; i < 2; i++)
    {
        rlen = cbuf_rlen(buffd);
        rbuf = (char *)cbuf_rptr(buffd);
        if(rlen > 0)
        {
            int ir = send(sockfd, rbuf, rlen, 0);
            
            //写满了?
            if(ir == -1 && errno == EAGAIN)
            {
                break;
            }
            //写满了
            if(ir == 0)
            {
                break;
            }

            if(ir > 0)
            {
                cbuf_rskip(buffd, ir);
            }
        }
    }
}
static void unix_socket_dispatch(int fd)
{
	int i;
	UnixSocket *unix_socket = fd2unix_socket(fd);
    int buffd = unix_socket->buf_read;
    for(;;)
    {
        js_sys("unix_socket", "dispatch\n");
        cbuf_info(buffd);

        int plen = cbuf_look_int(buffd);
        int buf_size = cbuf_size(buffd);
        js_sys("unix_socket", "plen :%d\n", plen);
        if(plen > 10000){
			char *rbuf = (char *)cbuf_base(buffd);
			for(i = 0; i < 50; i++){
				printf("%d\n", rbuf[i]);
			}
			sleep(1);
			int a = 1/0;
		}
        if(plen <= 0 || buf_size < plen)
        {
            js_sys("unix_socket", "port_dispatch is not finished\n");
            break;
        }
        char *rptr = (char *)cbuf_rptr(buffd);
        if(unix_socket->cb_packet)
        {
            int pktfd = pkt_new(buffd);
            int msgid = pkt2msgid(pktfd);
            if(unix_socket->cb_packet){
            	unix_socket->cb_packet(fd, msgid, pktfd);
            }
            pkt_free(pktfd);
        }else if(unix_socket->lua_cb_packet[0] != 0){

        	int pktfd = pkt_new(buffd);
        	int msgid = pkt2msgid(pktfd);
        	js_sys("unix_socket", "lua callback %s\n", unix_socket->lua_cb_packet);
        	luas_classfunc(unix_socket->lua_cb_self, unix_socket->lua_cb_packet, "(%d, %d, %d)",
        			fd, msgid, pktfd);
        	pkt_free(pktfd);
        }
        int size_read = buf_size - cbuf_size(buffd);
        //没读完的跳过
        cbuf_set_rptr(buffd, rptr);
        cbuf_rskip(buffd, plen);
        js_sys("unix_socket", "eat a packet\n");
    }
}
int unix_socket_close_line(int fd)
{

	UnixSocket *unix_socket = fd2unix_socket(fd);
	unix_socket->client_num -= 1;

    int sockfd = unix_socket->sockfd;

    //移动事件
    event_del(&unix_socket->ev_write);
    event_del(&unix_socket->ev_read);
    //释放buff
    cbuf_free(unix_socket->buf_read);
    cbuf_free(unix_socket->buf_write);
    //关闭socket
    close(sockfd);

    js_sys("unix_socket", "close line unix_socket_fd:%d, sockfd:%d\n", fd, sockfd);
}

static void unix_socket_on_read(int sockfd, short event, void *args)
{
	js_sys("unix_socket", "unix_socket_on_read\n");
    int i,j;
    int fd = (int)args;
    UnixSocket *unix_socket = fd2unix_socket(fd);
    int buffd = unix_socket->buf_read;

read_again:
    //保证buf还剩下这么多
    if(cbuf_remain(buffd) < 1024)
    {
        cbuf_double(buffd);
    }
    //一次至少能读到多少,这个数值关系到系统调用read的次数
    if(cbuf_wlen(buffd) < 64)
    {
        cbuf_2line(buffd);
    }
    cbuf_info(buffd);
    int   wlen = cbuf_wlen(buffd);
    char* wbuf = (char *)cbuf_wptr(buffd);
    js_sys("unix_socket", "goto read %d byte\n", wlen);
    int rt = recv(sockfd, wbuf, wlen, 0);
    js_sys("unix_socket", "system call read rt is:%d\n", rt);
    if(rt > 0)
    {
        cbuf_wskip(buffd, rt);
    }
    if(rt > 0 && rt < wlen)//读完了
    {
        js_sys("unix_socket", "port_on_read read finished\n");
        goto dispatch;
    }
    if(rt < 0 && errno == EAGAIN)//读完了
    {
        js_sys("unix_socket", "port_on_read read finished\n");
        goto dispatch;
    }
    if(rt > 0 && rt == wlen)//里面还有,继续读
    {
        js_sys("unix_socket", "port_on_read continue\n");
        goto dispatch;
    }
    if(rt == 0)
    {
        //close
        js_sys("unix_socket", "port_on_read close\n");
        close(sockfd);
        goto close;;
    }else //rt < 0
    {
        //error
        js_sys("unix_socket", "port_on_read error :%s\n", strerror(errno));
        goto close;;
    }
dispatch:
	cbuf_info(buffd);
	unix_socket_dispatch(fd);
    return;
close:
	unix_socket_close_line(fd);
    return;
}
static void unix_socket_on_accept(int listenfd, short event, void *args)
{
	js_sys("unix_socket", "accept a client\n");
	int fd = (int)args;
	UnixSocket* unix_socket = fd2unix_socket(fd);
	int sockfd;
    int portfd;
	struct sockaddr_un  addr;
	socklen_t addrlen = sizeof(addr);

    sockfd = accept(listenfd, (struct sockaddr*)&addr, &addrlen);

    unlink(addr.sun_path);

    if(unix_socket->client_num > 0){
    	close(sockfd);
    	js_sys("unix_socket", "one client allow\n");
    	return;
    }
    unix_socket->client_num += 1;
    setnonblock(sockfd);
        //初始化
    unix_socket->sockfd = sockfd;
    unix_socket->buf_read = cbuf_new(1024);
    unix_socket->buf_write = cbuf_new(1024);

    event_set(&unix_socket->ev_read,
        		unix_socket->sockfd,
                    EV_READ | EV_PERSIST,
                    unix_socket_on_read,
                    (void *)fd);
    event_set(&unix_socket->ev_write,
        		unix_socket->sockfd,
                    EV_WRITE | EV_PERSIST,
                    unix_socket_on_write,
                    (void *)fd);

    event_add(&unix_socket->ev_write, NULL);
    event_add(&unix_socket->ev_read, NULL);
}
int unix_socket_set_lua_cb_packet(int fd, int self, const char *funcname){
	js_sys("unix_socket", "unix_socket_set_lua_cb_packet %s\n", funcname);
	UnixSocket* unix_socket = fd2unix_socket(fd);
	strcpy(unix_socket->lua_cb_packet, funcname);
    
    unix_socket->lua_cb_self = self;
	return 1; 
}
int unix_socket_set_cb_packet(int fd, unix_socket_cb_packet cb)
{
	UnixSocket* unix_socket = fd2unix_socket(fd);
	unix_socket->cb_packet = cb;
    return 1;
}
int unix_socket_is_connectted(int fd){
	UnixSocket* unix_socket = fd2unix_socket(fd);
	return unix_socket->client_num >= 1;
}
void unix_socket_set_userdata(int fd, void *userdata)
{
	UnixSocket* unix_socket = fd2unix_socket(fd);
	unix_socket->userdata = userdata;
}
void *unix_socket_get_userdata(int fd)
{
	UnixSocket* unix_socket = fd2unix_socket(fd);
    return unix_socket->userdata;
}
int unix_socket_get_write_buffd(int fd){
	UnixSocket* unix_socket = fd2unix_socket(fd);
	return unix_socket->buf_write;
}
int unix_socket_listen(int fd, const char *path)
{
    int listenfd;
    int reuse = 1;
	struct sockaddr_un   addr;
	int addr_len;
    UnixSocket *unix_socket;

    unix_socket = fd2unix_socket(fd);
	listenfd = socket(AF_UNIX, SOCK_STREAM, 0);
	if(listenfd < 0)
	{
		js_sys("unix_socket", "failed to socket on file :%s\n", path);
		return 0;
	}
	unlink(path);
	bzero((void*)&addr, sizeof(addr));
	addr.sun_family  = AF_UNIX;
	strcpy(addr.sun_path, path);

	addr_len = offsetof(struct sockaddr_un, sun_path) + strlen(path);

		/* bind the name to the descriptor */
	if (bind(listenfd, (struct sockaddr *)&addr, addr_len) < 0) {
		js_sys("unix_socket", "fail to bind on file:%s\n", path);
		return 0;
	}

	if(listen(listenfd, 5) < 0)
	{
		js_sys("unix_socket", "fail to listen on file:%s\n", path);
		return 0;
	}

	unix_socket->listenfd = listenfd;

	event_set(&(unix_socket->evt),
			listenfd,
			EV_READ | EV_PERSIST,
			unix_socket_on_accept,
			(void*)fd);
	event_add(&(unix_socket->evt), NULL);
    js_sys("unix_socket", "listen on file:%s\n", path);
    return 1;
}


int unix_socket_connect(int fd, const char *path)
{

	struct sockaddr_un addr;
	int addr_len;

	UnixSocket *unix_socket;

	unix_socket = fd2unix_socket(fd);

	int sockfd = socket(AF_UNIX, SOCK_STREAM, 0);
	if(sockfd < 0)
	{
		js_sys("unix_socket", "create socket fail %s\n", path);
		return 0;
	}
	bzero((void*)&addr, sizeof(addr));
	addr.sun_family = AF_UNIX;
	strcpy(addr.sun_path, path);
	addr_len = offsetof(struct sockaddr_un, sun_path) + strlen(path);

	int rt = connect(sockfd, (struct sockaddr *)&addr, addr_len);
	if(rt < 0)
	{
		js_sys("unix_socket", "connect fail errno:%d:%s filepath:%s\n", errno, strerror(errno), path);
		return 0;
	}


	setnonblock(sockfd);
	//初始化
	unix_socket->sockfd = sockfd;
	unix_socket->buf_read = cbuf_new(1024);
	unix_socket->buf_write = cbuf_new(1024);

	event_set(&unix_socket->ev_read,
				unix_socket->sockfd,
				EV_READ | EV_PERSIST,
				unix_socket_on_read,
				(void *)fd);
	event_set(&unix_socket->ev_write,
				unix_socket->sockfd,
				EV_WRITE | EV_PERSIST,
				unix_socket_on_write,
				(void *)fd);

	event_add(&unix_socket->ev_write, NULL);
	event_add(&unix_socket->ev_read, NULL);
	return 1;
}
int unix_socket_send(int fd, unsigned int cmd, const char* buf, int buf_len)
{
    if(fd == 0){
        return 0;
    }
    UnixSocket *unix_socket = fd2unix_socket(fd);
    int buffd = unix_socket->buf_write;
    cbuf_write_int(buffd, buf_len + 8);
    cbuf_write_int(buffd, cmd);
    cbuf_write(buffd, buf, buf_len);

    cbuf_info(buffd);
    return 1;
}
static int New(lua_State *L)
{
    if (LUA_ARGNUM == 0)
    {
    	int rt = unix_socket_new();
        LUA_RETURN(rt);
    }else{
    	js_sys("unix_socket", "interface Listen arg invalid\n");
    }

    LUA_RETURN(0);
}
static int Connect(lua_State *L)
{
    if (LUA_ARGNUM == 2 && LUA_ISNUMBER(1) && LUA_ISSTRING(2))
    {
        int fd = (int)LUA_TONUMBER(1);
        const char *filepath = (const char *)LUA_TOSTRING(2);
        int rt = unix_socket_connect(fd, filepath);
        LUA_RETURN(rt);
    }

    LUA_RETURN(0);
}
static int OnPacket(lua_State *L)
{
	js_sys("unix_socket", "OnPacket\n");
    if (LUA_ARGNUM == 3 && LUA_ISNUMBER(1) && LUA_ISSTRING(2))
    {
        int fd                  = (int)LUA_TONUMBER(1);
        const char *funcname    = (const char *)LUA_TOSTRING(2);
        int self                = (int)luaL_ref(L, LUA_REGISTRYINDEX);
        
        int rt = unix_socket_set_lua_cb_packet(fd, self, funcname);
        LUA_RETURN(rt);
    }

    LUA_RETURN(0);
}
static int Listen(lua_State *L)
{
    if (LUA_ARGNUM == 2 && LUA_ISNUMBER(1) && LUA_ISSTRING(2))
    {
        int fd = (int)LUA_TONUMBER(1);
        const char *filepath = (const char *)LUA_TOSTRING(2);
        int rt = unix_socket_listen(fd, filepath);
        LUA_RETURN(rt);
    }else{
    	js_sys("unix_socket", "interface Listen arg invalid\n");
    }

    LUA_RETURN(0);
}

static int Send(lua_State *L)
{
    int i;
	int arg_num = LUA_ARGNUM;


	if (arg_num >= 2 && LUA_ISNUMBER(1) && LUA_ISNUMBER(2) )
	{
		int buffd = cbuf_new(1024);
        int fd = (int)LUA_TONUMBER(1);

        unsigned int cmd = (unsigned int) LUA_TONUMBER(2);


        for (i = 3; i <= arg_num; i++)
        {
            if (LUA_ISNUMBER(i))
            {
                cbuf_write_int(buffd, (int) LUA_TONUMBER(i));
            }

            else if (LUA_ISSTRING(i))
            {
                cbuf_write_utf(buffd, LUA_TOSTRING(i));
            }
        }
        unix_socket_send(fd, cmd, (const char *)cbuf_base(buffd), cbuf_size(buffd));
        cbuf_free(buffd);
	}

    LUA_RETURN(1);
}
static luaL_Reg lua_lib[] =
{
	LUA_FUNC_INIT(New),
	LUA_FUNC_INIT(Send),
    LUA_FUNC_INIT(Listen),
    LUA_FUNC_INIT(Connect),
    LUA_FUNC_INIT(OnPacket),

    {NULL, NULL}
};

int unix_socket_init()
{
	luas_reg_lib("UnixSocket", lua_lib);

}
